package com.atguigu.func;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author Felix
 * @date 2024/7/13
 */
public class MySinkFunction extends RichSinkFunction<WaterSensor> {
    private  Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8", "root", "123456");
    }

    @Override
    public void close() throws Exception {
        if(conn != null){
            conn.close();
        }
    }

    @Override
    public void invoke(WaterSensor ws, Context context) throws Exception {
       //创建数据库操作对象
        PreparedStatement ps = conn.prepareStatement("insert into ws values(?,?,?)");
        ps.setString(1,ws.getId());
        ps.setLong(2,ws.getTs());
        ps.setString(3,ws.getVc() + "");
        //执行SQL语句
        ps.execute();
        //释放资源
        if(ps != null){
            ps.close();
        }
    }
}
